package com.hua.api.modules.home.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.hua.api.core.activemq.ActiveMqConstant;
import com.hua.api.core.activemq.MqMessageBuilder;
import com.hua.api.core.activemq.consumer.event.EventConstant;
import com.hua.api.core.constants.Status;
import com.hua.api.modules.home.service.ActiveMqService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import javax.jms.*;
import java.util.Set;


@Slf4j
@RequiredArgsConstructor
@Service
public class ActiveMqServiceImpl implements ActiveMqService {


    public static final  int NON_PERSISTENT = 1;

    public static final  int PERSISTENT = 2;

    private final JmsTemplate jmsQueueTemplate;

    private final JmsTemplate jmsTopicTemplate;


    @Override
    public void onUserMesssage(Set<Integer> set,String title, String context,String type,Integer needPush,String soundPath) {
        if(set.size()>0){
            JSONArray array = new JSONArray();
            for(Integer userId:set){
                array.add(userId);
            }

            JSONObject msgData = MqMessageBuilder.builder()
                    .target(ActiveMqConstant.hua000_api_queue)
                    .event(EventConstant.event_user_message)
                    .set("userIds",array)
                    .set("type",type)
                    .set("title",title)
                    .set("context",context)
                    .set("needPush",needPush)
                    .set("soundPath",soundPath)
                    .build();

            sendQueue(ActiveMqConstant.hua000_api_queue,msgData.toJSONString());
        }
    }

    @Override
    public void onOrderInfoLogger(Integer memberId,Integer userId,Integer orderId,String operateEvent,String operateName,String operateDesc) {

        JSONObject msgData = MqMessageBuilder.builder()
                .target(ActiveMqConstant.hua000_api_queue)
                .event(EventConstant.event_order_logger)
                .set("memberId",memberId)
                .set("userId",userId)
                .set("orderId",orderId)
                .set("operateEvent",operateEvent)
                .set("operateName",operateName)
                .set("operateDesc",operateDesc)
                .build();

        sendQueue(ActiveMqConstant.hua000_api_queue,msgData.toJSONString());
    }


    @Override
    public void onSmsSender(Set<String> set,String context){
        if(set.size()>0){
            JSONArray array = new JSONArray();
            for(String phoneNumber:set){
                array.add(phoneNumber);
            }
            JSONObject msgData = MqMessageBuilder.builder()
                    .target(ActiveMqConstant.hua000_api_queue)
                    .event(EventConstant.event_sms_sender)
                    .set("phoneNumbers",array)
                    .set("context",context)
                    .build();
            sendQueue(ActiveMqConstant.hua000_api_queue,msgData.toJSONString());
        }
    }


    @Override
    public void onOrderBidCreate(Integer bidId) {
        JSONObject data = new JSONObject();
        data.put("bidId",bidId);
        data.put("action","create");
        data.put("timestamp",System.currentTimeMillis());
        JSONObject msgData = MqMessageBuilder.builder()
                .target(ActiveMqConstant.hua000_store_queue)
                .event(EventConstant.event_order_bid)
                .data(data)
                .build();
        sendQueue(ActiveMqConstant.hua000_store_queue,msgData.toString());
    }

    @Override
    public void sendQueue(String destination, String message) {
        log.info("ActiveMqServiceImpl.sendQueue(destination={},message={})",destination,message);
        ActiveMQQueue queue = new ActiveMQQueue(destination);
        jmsQueueTemplate.convertAndSend(queue,message);
    }

    @Override
    public void sendDelayQueue(String destination, String message, Long time) {
        log.info("ActiveMqServiceImpl.senDelayQueue(destination={},message={},current={})",destination,message,System.currentTimeMillis());
        ActiveMQQueue queue = new ActiveMQQueue(destination);
        delaySendMessage(queue,message,time);
    }

    @Override
    public void sendTopic(String destination, String message) {
        log.info("ActiveMqServiceImpl.sendTopic(destination={},message={},current={})",destination,message,System.currentTimeMillis());
        ActiveMQTopic topic = new ActiveMQTopic(destination);
        jmsTopicTemplate.convertAndSend(topic,message);
    }

    @Override
    public void sendDelayTopic(String destination, String message, Long time) {
        log.info("ActiveMqServiceImpl.sendDelayTopic(destination={},message={},current={})",destination,message,System.currentTimeMillis());
        ActiveMQTopic topic = new ActiveMQTopic(destination);
        delaySendMessage(topic,message,time);
    }

    private void delaySendMessage(Destination destination, String data, Long time){
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;

        ConnectionFactory connectionFactory = jmsQueueTemplate.getConnectionFactory();
        try{
            connection = connectionFactory.createConnection();
            connection.start();
            // 获取session，true开启事务，false关闭事务
            session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

            producer = session.createProducer(destination);

            //消息持久化
            producer.setDeliveryMode(PERSISTENT);
            ObjectMessage message = session.createObjectMessage(data);
            //设置延迟时间
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
            // 发送消息
            producer.send(message);
            //一旦开启事务发送，那么就必须使用commit方法提交事务
            session.commit();

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                if (producer != null){
                    producer.close();
                }
                if (session != null){
                    session.close();
                }
                if (connection != null){
                    connection.close();
                }
            } catch (Exception e){
                e.printStackTrace();
            }
        }


    }
}
